package com.bwie.realtime.jtp.dwd.log.gd;

import net.bwie.realtime.jtp.utils.KafkaUtil;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class AppDWD {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
//        DataStream<String> dwd_traffic_is_new = KafkaUtil.consumerKafka(env, "dwd_traffic_is_new");
//        dwd_traffic_is_new.print();
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        /*

        CREATE TABLE KafkaTable (
            common map<string,string>,
            page map<string,string>,
            ts bigint
             time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),
                 WATERMARK FOR time_ltz AS time_ltz - INTERVAL '0' SECOND
) WITH (
  'connector' = 'kafka',
  'topic' = 'dwd-traffic-error-log',
  'properties.bootstrap.servers' = 'node101:9092',
  'properties.group.id' = 'testGroup',
  'scan.startup.mode' = 'earliest-offset',
  'format' = 'json'
)
            create table app_log_page(

            )
         */
        tenv.executeSql(" CREATE TABLE k1 (\n" +
                "            common map<string,string>,\n" +
                "            page map<string,string>,\n" +
                "            ts bigint,\n" +
                "             time_ltz AS TO_TIMESTAMP_LTZ(ts, 3),\n" +
                "                 WATERMARK FOR time_ltz AS time_ltz - INTERVAL '0' SECOND\n" +
                ") WITH (\n" +
                "  'connector' = 'kafka',\n" +
                "  'topic' = 'dwd-traffic-error-log',\n" +
                "  'properties.bootstrap.servers' = 'node101:9092',\n" +
                "  'properties.group.id' = 'testGroup',\n" +
                "  'scan.startup.mode' = 'earliest-offset',\n" +
                "  'format' = 'json'\n" +
                ")");
        tenv.sqlQuery("select page[item_type] from k1 group by page[item_type]").execute().print();


        env.execute("aa");
    }
}
